package org.example.MysqlToHive;


import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MysqlToHive {
    // MySQL到Hive数据类型映射
    private static final Map<String, String> TYPE_MAPPING = new HashMap<>();
    static {
        TYPE_MAPPING.put("tinyint", "TINYINT");
        TYPE_MAPPING.put("smallint", "SMALLINT");
        TYPE_MAPPING.put("mediumint", "INT");
        TYPE_MAPPING.put("int", "INT");
        TYPE_MAPPING.put("bigint", "INT");
        TYPE_MAPPING.put("float", "FLOAT");
        TYPE_MAPPING.put("double", "DOUBLE");
        TYPE_MAPPING.put("decimal", "DECIMAL");
        TYPE_MAPPING.put("char", "STRING");
        TYPE_MAPPING.put("varchar", "STRING");
        TYPE_MAPPING.put("text", "STRING");
        TYPE_MAPPING.put("date", "DATE");
        TYPE_MAPPING.put("datetime", "TIMESTAMP");
        TYPE_MAPPING.put("timestamp", "TIMESTAMP");
        TYPE_MAPPING.put("time", "STRING");
        TYPE_MAPPING.put("boolean", "BOOLEAN");
        TYPE_MAPPING.put("blob", "BINARY");
    }

    public static void main(String[] args) {

//        String mysqlJdbcUrl = "jdbc:mysql://10.10.10.164:3306/ppm_plan";
//        String user = "ppment";
//        String password = "Pr0PlanMent@6789";
//        String tableName = "t_disclose";
//        String databaseName = "ppm_plan";

        String mysqlJdbcUrl = "";
        String user = "";
        String password = "";
        String tableName = "";
        String databaseName = "";

        try {
            List<String> lines = Files.readAllLines(Paths.get("D:\\code\\MySQLToHive\\src\\main\\java\\a.txt"), Charset.forName("UTF-8"));
            for (String line : lines) {
                if(line.contains("url")){
                    mysqlJdbcUrl = line.split("=")[1];
                }
                if(line.contains("user")){
                    user = line.split("=")[1];
                }
                if(line.contains("password")){
                    password = line.split("=")[1];
                }
                if(line.contains("databaseName")){
                    databaseName = line.split("=")[1];
                }
                if(line.contains("tableName")){
                    tableName = line.split("=")[1];
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        String[] tables = tableName.split(",");
        for (int length = tables.length -1 ; length >= 0; length--) {
            try {
                System.out.println();

                String hiveDDL = convert(mysqlJdbcUrl, user, password,databaseName,tables[length]);
                System.out.println(hiveDDL);

                System.out.println();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    public static String convert(String jdbcUrl, String user, String password,String databaseName ,String tableName)
            throws SQLException {

        StringBuilder ddl = new StringBuilder();

        ddl.append("CREATE TABLE IF NOT EXISTS ods.ods_mysql_" +databaseName+ "_"+ tableName + "_full_1d ( \n");

        Connection conn = DriverManager.getConnection(jdbcUrl, user, password);
        // 创建Statement对象
        Statement stmt = conn.createStatement();

        String sql = "SELECT table_comment FROM information_schema.TABLES WHERE table_schema = " + "'" + databaseName + "'" +" and table_name =" + "'" + tableName + "'"  ;
        // 执行查询语句

        ResultSet tabRs = stmt.executeQuery(sql);
        String tN = "";
        while (tabRs.next()) {
            tN =  tabRs.getString("table_comment");
        }
        tabRs.close();
        String sql1 = "SHOW FULL COLUMNS FROM " + tableName + ";";
        ResultSet rs = stmt.executeQuery(sql1);
        // 处理结果集
        while (rs.next()) {

            String type = "";
            String last = "";
            if(rs.getString("Type").contains("(")){
                type= rs.getString("Type").substring(0, rs.getString("Type").lastIndexOf("(") );
                if (TYPE_MAPPING.get(type) == "FLOAT" || TYPE_MAPPING.get(type) == "DOUBLE" || TYPE_MAPPING.get(type) == "DECIMAL") {
                    last = rs.getString("Type").substring(rs.getString("Type").lastIndexOf("("));
                }else {
                    last = "";
                }
            }else{
                type= rs.getString("Type");
            };
            ddl.append(rs.getString("Field"))
                    .append("\t")
                    .append(
                            TYPE_MAPPING.get(type)
                    )
                    .append(last)
                    .append("\tcomment\t'")
                    .append(rs.getString("Comment"))
                    .append("',\n");
            type = null;
            last = null;
        }
        rs.close();
        ddl = new StringBuilder(ddl.substring(0, ddl.length() - 2));
        ddl.append(") \ncomment\t\"").append(tN).append("\"\n").append("stored as orc\n;");
        // 添加存储格式和位置（按需修改）
        tN="";
        return ddl.toString();
    }

}
